#include "MessageCenter.h"
#include "MessageObserver.h"
#include "Message.h"
#include "MessagePoster.h"
#include <sys/time.h>


static MessageCenter *_sharedMessageCenter;


//
// Initialize a new MessageCenter object.
//
MessageCenter::MessageCenter()
{
	//
	// Initialize the mutexes and conditions.
	//
	pthread_mutex_init(&_observersMutex, NULL);
	pthread_mutex_init(&_messagesMutex, NULL);
	pthread_cond_init(&_postCondition, NULL);

	//
	// Initialize the maps.
	//
	_observers = std::map<std::string, std::list<MessageObserver *> >();
	_messages = std::map<pthread_t, std::list<MessagePoster *> >();
}


//
// Process all the messages in the list of messages. Each message is delete'd from
// memory after it is processed.
//
void MessageCenter::ProcessMessages(std::list<MessagePoster *> &messages)
{
	std::list<MessagePoster *>::iterator it;
	MessagePoster *poster;
	
	
	while ((it = messages.begin()) != messages.end())
	{
		//
		// Remove the message from the list.
		//
		poster = *it;
		messages.erase(it);

		//
		// Process the message.
		//
		ProcessMessage(poster);
		
		//
		// Delete the Message object.
		//
		delete poster->_message;
		delete poster;
	}
}


//
// Process a single message by calling it's callback function.
//
void MessageCenter::ProcessMessage(MessagePoster *poster)
{
	poster->_callback(poster->_message, poster->_userInfo);
}


//
// Retrieve the shared message center for this process.
//
MessageCenter &MessageCenter::SharedMessageCenter()
{
	if (_sharedMessageCenter == NULL)
		_sharedMessageCenter = new MessageCenter();

	return *_sharedMessageCenter;
}


//
// Add a new observer to the message center. The observer is the current thread
// and it will be observing the given message name.
//
void MessageCenter::AddObserver(std::string name, void (*callback)(Message *, void *), void *userInfo)
{
	std::list<MessageObserver *> *observers;
	std::list<MessageObserver *>::iterator it;
	pthread_t thread_id = pthread_self();
	
	
	//
	// Lock the mutex to prevent changes and then get the list of observing
	// threads for this message name.
	//
	pthread_mutex_lock(&_observersMutex);
	observers = &_observers[name];
	
	//
	// See if this thread is already observing this message name.
	//
	for (it = observers->begin(); it != observers->end(); it++)
	{
		if ((*it)->_thread_id == thread_id)
			break;
	}
	
	//
	// If an existing thread_id was not found, then add it.
	//
	if (it == observers->end())
	{
		observers->push_back(new MessageObserver(thread_id, name, callback, userInfo));
	}
		
	//
	// Unlock the mutex.
	//
	pthread_mutex_unlock(&_observersMutex);
}


//
// A thread is no longer interested in observing the given message name.
//
void MessageCenter::RemoveObserver(std::string name)
{
	std::list<MessageObserver *> *observers;
	std::list<MessageObserver *>::iterator it;
	pthread_t thread_id = pthread_self();
	
	
	//
	// Lock the mutex to prevent changes and then get the list of observing
	// threads for this message name.
	//
	pthread_mutex_lock(&_observersMutex);
	observers = &_observers[name];
	
	//
	// See if this thread is observing this message name and if so remove it.
	//
	for (it = observers->begin(); it != observers->end(); it++)
	{
		if ((*it)->_thread_id == thread_id)
		{
			MessageObserver *observer;

			observer = *it;
			observers->erase(it);
			delete observer;

			break;
		}
	}
	
	//
	// Unlock the mutex.
	//
	pthread_mutex_unlock(&_observersMutex);
}


//
// Post the given message into the message queues of all observing threads.
//
void MessageCenter::PostMessage(Message msg)
{
	std::list<MessageObserver *> *observers;
	std::list<MessageObserver *> postThreads = std::list<MessageObserver *>();
	std::list<MessageObserver *>::iterator it;


	//
	// Lock out other threads from messing with the observer list and generate a
	// temporary list of which threads we will post to.
	//
	pthread_mutex_lock(&_observersMutex);
	observers = &_observers[msg.getName()];
	for (it = observers->begin(); it != observers->end(); it++)
	{
		postThreads.push_back(*it);
	}
	pthread_mutex_unlock(&_observersMutex);

	//
	// Now lock out threads from accessing the list of messages and post a copy
	// of the message to each thread's queue.
	//
	pthread_mutex_lock(&_messagesMutex);
	for (it = postThreads.begin(); it != postThreads.end(); it++)
	{
		(&_messages[(*it)->_thread_id])->push_back(new MessagePoster(msg.copy(), (*it)->_callback, (*it)->_userInfo));
	}
	pthread_mutex_unlock(&_messagesMutex);

	//
	// If we posted the message to any queues, broadcast.
	//
	if (postThreads.size() > 0)
		pthread_cond_broadcast(&_postCondition);
}


//
// Wait forever for a message to be received. Before this method starts waiting it
// will process all pending messages. This allows the caller to do any other processing.
// If the method processed any messages before waiting then it returns immediately.
//
void MessageCenter::WaitForMessage()
{
	WaitForMessage(-1);
}


//
// This message is the same as WaitForMessage() except that it will only wait for mstimeout
// milliseconds before it returns.
//
void MessageCenter::WaitForMessage(long mstimeout)
{
	std::list<MessagePoster *> *messages, myMessages = std::list<MessagePoster *>();
	std::list<MessagePoster *>::iterator it;
	pthread_t thread_id = pthread_self();
	bool foundMessages = false;
	
	
	//
	// Check for any existing messages in our queue.
	//
	pthread_mutex_lock(&_messagesMutex);
	messages = &_messages[thread_id];
	while ((it = messages->begin()) != messages->end())
	{
		myMessages.push_back(*it);
		messages->erase(it);
		foundMessages = true;
	}
	pthread_mutex_unlock(&_messagesMutex);
	
	//
	// Process any messages we found.
	//
	ProcessMessages(myMessages);
	
	//
	// Skip waiting for more messages if we already processed some.
	//
	if (foundMessages)
		return;

	//
	// Wait for the next message.
	//
	pthread_mutex_lock(&_messagesMutex);
	if (mstimeout >= 0)
	{
		struct timeval tv;
		struct timespec ts;
		gettimeofday(&tv, NULL);
		ts.tv_sec = tv.tv_sec;
		ts.tv_nsec = (tv.tv_usec * 1000);

		//
		// Add the wait time.
		//
		ts.tv_sec += (mstimeout / 1000);
		mstimeout %= 1000;
		ts.tv_nsec += (mstimeout * 1000000);
		if (ts.tv_nsec > 1000000000)
		{
			ts.tv_sec += (ts.tv_nsec / 1000000000);
			ts.tv_nsec %= 1000000000;
		}

		pthread_cond_timedwait(&_postCondition, &_messagesMutex, &ts);
	}
	else
		pthread_cond_wait(&_postCondition, &_messagesMutex);
	messages = &_messages[thread_id];
	while ((it = messages->begin()) != messages->end())
	{
		myMessages.push_back(*it);
		messages->erase(it);
	}
	pthread_mutex_unlock(&_messagesMutex);

	//
	// Process any final messages we got.
	//
	ProcessMessages(myMessages);
}

